package com.hup.utils.io.netty;

import com.hup.utils.commons.exception.AppExceptionHandler;
import com.hup.utils.io.netty.NettyTCPServer.ServerConfig;
import com.hup.utils.io.netty.handler.*;
import com.hup.utils.io.netty.handler.DataHandler.DataMethodInvoker;
import com.hup.utils.io.netty.handler.HeartbeatHandler.IdleCallback;
import com.hup.utils.io.netty.model.BaseDataPacket;
import com.hup.utils.io.netty.model.LoginRequestPacket;
import com.hup.utils.io.netty.model.LoginResponsePacket;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.log4j.Log4j2;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * netty,nio,TCP客户端;<pre>
 * 客户端的线程池:
 * .只在 {@link #NettyTCPClient(ClientConfig)}构造时进行初始化,然后就不能修改了
 * .在{@link #doTerminate()}时才会释放,因为相当于服务端,客户端会比更频繁的连接和断开,期间不需要修改线程池策略
 * .{@link .#doTerminate()} 销毁后,本实例就不能重复使用! 因为 {@link ConnectedObj#releaseExecutor()}
 *
 * @author hugan
 * @date 2021/4/25
 */
@Log4j2
public class NettyTCPClient {
    private final ClientConfig clientConfig;
    private final ConnectedObj connectedObj;
    private final ClientHandler clientHandlerInstance;

    public NettyTCPClient(ClientConfig clientConfig) {
        this.clientConfig = clientConfig;
        connectedObj = new ConnectedObj();
        connectedObj.buildExecutor(clientConfig);
        connectedObj.dataHandlerMap = clientConfig.registerDataHandlerMap();
        clientHandlerInstance = new ClientHandler();
    }

    /**
     * {@link Bootstrap#connect(String, int)}
     */
    public void connect() {
        connectedObj.assertState(ConnectState.DISCONNECT);
        log.info("正在连接服务端");
        connectedObj.connectState = ConnectState.CONNECTING;

        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(connectedObj.workerGroup)//指定线程模型
                .channel(NioSocketChannel.class) //指定IO类型为NIO
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel c) {
                        c.pipeline().addLast(new DataSplitter());
                        c.pipeline().addLast(DataCodecHandler.INSTANCE);
                        if (clientConfig.loginHandler != null) c.pipeline().addLast(clientConfig.loginHandler);
                        if (clientConfig.addHeartbeat) HeartbeatHandler.addHeartbeatHandler(c.pipeline());
                        c.pipeline().addLast(clientHandlerInstance);
                    }
                });
        if (clientConfig.connectTimeout != null)
            bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, clientConfig.connectTimeout);
        bootstrap.connect(clientConfig.host, clientConfig.port).addListener(future -> {
            boolean success = future.isSuccess();
            log.info("connect.res={}", success);
            if (!success || clientConfig.loginHandler == null) {
                //连接失败,或无认证功能,则马上通知callback
                connectedObj.connectState = success ? ConnectState.CONNECTED : ConnectState.DISCONNECT;
                //连接成功:无认证功能,所以loginResponse为null;连接失败:socket无响应
                LoginResponsePacket loginResponse = success ? null : LoginResponsePacket.failure("服务端无响应");
                clientConfig.callback.onConnectResult(success, clientConfig.host, clientConfig.port, loginResponse);
            }
            //else 否则,如果服务端正常回复,则在channelRead0收到LoginResponsePacket后通知callback
            //此处获取不了context,只能在后面紧接的channelActive回调中获取(是同一条线程)
        });
    }

    public void disconnect() {
        connectedObj.assertState(ConnectState.CONNECTED);
        log.info("正在断开连接");
        connectedObj.clientContext.close().addListener(future -> clientHandlerInstance.onClientDisconnectResult(future.isSuccess()));
    }

    public ConnectState getConnectState() {
        return connectedObj.connectState;
    }

    public void send(BaseDataPacket data) {
        connectedObj.assertState(ConnectState.CONNECTED);
        connectedObj.clientContext.writeAndFlush(data);
    }

    /**
     * 注意: {@link NettyTCPClient}
     */
    public void doTerminate() {
        if (ConnectState.CONNECTED.equals(connectedObj.connectState)) {
            connectedObj.clientContext.close().addListener(future -> {
                clientHandlerInstance.onClientDisconnectResult(future.isSuccess());
                connectedObj.releaseExecutor();
            });
        } else {
            connectedObj.releaseExecutor();
        }
    }

    @Sharable
    private class ClientHandler extends SimpleChannelInboundHandler<BaseDataPacket> {

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            log.info("");
            connectedObj.clientContext = ctx;
            if (clientConfig.loginHandler != null) {
                LoginRequestPacket loginValidData = clientConfig.loginHandler.getLoginValidData();
                log.info("发送登陆请求, 当前连接状态:{}", connectedObj.connectState.desc);//仍然是连接中
                ctx.channel().writeAndFlush(loginValidData);
                ctx.channel().eventLoop().schedule(() -> {
                    Channel channel = ctx.channel();
                    LoginResponsePacket loginResponse = clientConfig.loginHandler.getLoginResponseData(channel);
                    if (loginResponse == null) {
                        log.warn("服务端无回复认证请求,主动断开连接:{}", channel.remoteAddress());
                        channel.close();
                    }
                }, clientConfig.getLoginTimeout(), TimeUnit.MILLISECONDS);
            }
            super.channelActive(ctx);
        }

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, BaseDataPacket msg) throws Exception {
            if (clientConfig.loginHandler != null && msg instanceof LoginResponsePacket) {
                log.debug("收到登陆回复:{}", msg);
                //由LoginResponseHandler传递的消息,收到服务端认证回复
                LoginResponsePacket loginResponse = (LoginResponsePacket) msg;
                if (loginResponse.isSuccess()) {
                    connectedObj.connectState = ConnectState.CONNECTED;
                    clientConfig.callback.onConnectResult(true, clientConfig.host, clientConfig.port, loginResponse);
                } else {
                    /*
                    认证失败时server会主动断开,然后到达channelInactive
                    这里为了安全(避免连接了其他程序),也主动close
                     */
                    ctx.channel().close();
                }
                return;
            }
            DataMethodInvoker invoker = connectedObj.dataHandlerMap.get(msg.getClass());
            if (invoker != null) {
                //已指定处理方法的数据:通过ServerConfig.dataHandler配置指定的
                try {
                    //log.info("mgs={}", msg);
                    invoker.invoke(msg, ctx.channel());
                } catch (Exception e) {
                    log.error("消息处理异常={}", msg);
                    AppExceptionHandler.handle("消息处理异常", e);
                }
            } else {
                //未知数据/未指定处理方法的数据
                clientConfig.callback.onUnknownData(msg);
            }
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            connectedObj.connectState = ConnectState.DISCONNECT;
            connectedObj.clientContext = null;
            Channel channel = ctx.channel();
            if (clientConfig.loginHandler == null || LoginResponseHandler.isLoginResponseSuccess(channel)) {
                //没有登陆认证功能,或已经通过认证,则断线后通知callback;
                log.info("连接断开={}", channel.remoteAddress());
                clientConfig.callback.onConnectionLost();
            } else {
                log.warn("未通过认证的连接断开={}", channel.remoteAddress());
                LoginResponsePacket loginResponse = clientConfig.loginHandler.getLoginResponseData(channel);
                if (loginResponse == null) loginResponse = LoginResponsePacket.failure("服务端无回复认证请求");
                clientConfig.callback.onConnectResult(false, clientConfig.host, clientConfig.port, loginResponse);
            }
            super.channelInactive(ctx);
        }

        private void onClientDisconnectResult(boolean success) {
            if (!success) {
                log.error("连接断开失败??");
            }
            //else channelInactive 会被调用
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable e) throws Exception {
            clientConfig.callback.exceptionCaught(ctx, e);
        }
    }

    /**
     * 客户端连接后的状态对象
     */
    public static class ConnectedObj {
        private ConnectState connectState = ConnectState.DISCONNECT;
        private NioEventLoopGroup workerGroup;
        private ChannelHandlerContext clientContext;
        private Map<Class<? extends BaseDataPacket>, DataMethodInvoker> dataHandlerMap;

        private void assertState(ConnectState state) {
            if (connectState != state) throw new IllegalArgumentException("连接状态为:" + connectState.desc);
        }

        private void buildExecutor(ClientConfig clientConfig) {
            if (clientConfig.nioEventLoopGroupBuilder != null) {
                workerGroup = clientConfig.nioEventLoopGroupBuilder.get(true);
            } else {
                workerGroup = new NioEventLoopGroup();
            }
        }

        private void releaseExecutor() {
            workerGroup.shutdownGracefully();
            workerGroup = null;
        }
    }

    /**
     * 客户端连接配置类;
     * 配置在下一次 {@link NettyTCPClient#connect()}时生效
     */
    public static class ClientConfig implements DataHandlerRegister {

        private String host;
        private int port;
        /**
         * 连接超时,单位ms
         */
        @Setter
        private Integer connectTimeout;
        private ClientCallback callback;
        private LoginResponseHandler<?, ?> loginHandler;
        /**
         * 发送登陆认证请求后,n毫秒内没收到服务端的回复,则主动断开连接;
         * 有登陆认证时业务时才使用
         */
        @Setter
        @Getter
        private int loginTimeout = 1000;
        private boolean addHeartbeat;
        /**
         * 数据处理类,每个元素相当于一个Controller
         */
        @Getter
        private ArrayList<DataHandler> lstDataHandler = new ArrayList<>();
        /**
         * 配置自定义的线程池,注意 {@link NettyTCPClient}
         */
        @Setter
        private NioEventLoopGroupBuilder nioEventLoopGroupBuilder;

        public ClientConfig(ClientCallback callback) {
            this.callback = callback;
        }

        public ClientConfig target(String host, int port) {
            this.host = host;
            this.port = port;
            return this;
        }

        /**
         * 配置登陆认证功能;
         * [登陆认证功能]双向配置流程:
         * .服务端初始化时,调用 {@link ServerConfig#loginHandler(LoginHandler)}
         * .客户端调用本方法
         *
         * @param loginValidCallback null-移除登录认证功能;notNull-添加登录认证功能
         */
        public ClientConfig loginHandler(LoginResponseHandler<?, ?> loginValidCallback) {
            this.loginHandler = loginValidCallback;
            LoginResponseHandler.handleLoginResponseCodec(loginValidCallback);
            return this;
        }

        /**
         * 配置有无心态机制; {@link HeartbeatHandler}
         */
        public ClientConfig setAddHeartbeat(boolean addHeartbeat, IdleCallback idleCallback) {
            return setAddHeartbeat(addHeartbeat, idleCallback, HeartbeatHandler.getReaderIdleTimeSeconds());
        }

        /**
         * @param readerIdleTimeSeconds {@link HeartbeatHandler#getReaderIdleTimeSeconds()}
         */
        public ClientConfig setAddHeartbeat(boolean addHeartbeat, IdleCallback idleCallback, int readerIdleTimeSeconds) {
            this.addHeartbeat = addHeartbeat;
            if (addHeartbeat && idleCallback == null) throw new IllegalArgumentException("idleCallback不能为空");
            HeartbeatHandler.initHeartBeat(true, addHeartbeat ? idleCallback : null, readerIdleTimeSeconds);
            return this;
        }

    }

    /**
     * 客户端启动状态
     */
    public enum ConnectState {
        DISCONNECT("已断开"),
        CONNECTED("已连接"),
        CONNECTING("连接中");
        private final String desc;

        ConnectState(String desc) {
            this.desc = desc;
        }
    }

    /**
     * 客户端状态回调
     */
    public interface ClientCallback {

        /**
         * 连接服务端结果回调
         *
         * @param host 服务端host
         * @param port 服务端port
         * @param loginResponse 登陆认证回复,有认证功能时才不为null,数据类型是绑定 {@link LoginResponseHandler}时配置的
         */
        void onConnectResult(boolean success, String host, Integer port, LoginResponsePacket loginResponse);

        /**
         * 连接已断开
         */
        void onConnectionLost();

        /**
         * 收到客户端的未知消息
         */
        default void onUnknownData(BaseDataPacket data) {
            log.warn("未知数据类型={}", data);
        }

        default void exceptionCaught(ChannelHandlerContext ctx, Throwable e) throws Exception {
            if (e instanceof IOException) {
                /*
                代码没bug时,一般是server程序非正常结束,没有正常关闭socket
                    导致IO异常:远程主机强迫关闭了一个现有的连接
                    此时onConnectionLost已经通知callback了,所以不需要打印
                注意:如果是其他问题,如代码bug引起的IO异常,则此处是一个大坑,会屏蔽异常信息,
                    但代码规范是,实现业务逻辑时,应自己捕获IO异常!
                 */
                return;
            }
            log.error("channel={}", ctx.channel().remoteAddress(), e);
        }
    }

}
